package ex.datastream.watermark;

import ex.datatypes.TaxiRide;
import ex.util.generator.TaxiRideGeneratorFromMysql;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.time.Duration;

/**
 * https://blog.csdn.net/qq_34125999/article/details/128971255
 *
 * select * from taxi_ride where driverId in('2013000108','2013000134','2013000062') order by rideId limit 10
 *
 * forBoundedOutOfOrderness(Duration.ofMinutes(10))
 *
 2	1	2020-01-01 20:00:40	-73.856	40.7741	-73.802	40.8429	3	2013000108	2013000108         ----->[Y]
 2	0	2020-01-01 20:31:40	-73.856	40.7741	-73.802	40.8429	3	2013000108	2013000108
 3	1	2020-01-01 20:01:00	-73.8645	40.7633	-73.848	40.7844	3	2013000134	2013000134     ----->[Y]
 3	0	2020-01-01 20:27:00	-73.8645	40.7633	-73.848	40.7844	3	2013000134	2013000134
 4	1	2020-01-01 20:01:20	-73.8609	40.7679	-73.7818	40.8686	3	2013000062	2013000062 ----->[Y]
 4	0	2020-01-01 20:29:20	-73.8609	40.7679	-73.7818	40.8686	3	2013000062	2013000062
 243	1	2020-01-01 21:21:00	-73.867	40.7601	-73.7953	40.8515	3	2013000108	2013000108
 243	0	2020-01-01 21:46:00	-73.867	40.7601	-73.7953	40.8515	3	2013000108	2013000108
 289	1	2020-01-01 21:36:20	-73.7677	40.8866	-73.8502	40.7816	3	2013000062	2013000062
 289	0	2020-01-01 22:58:20	-73.7677	40.8866	-73.8502	40.7816	3	2013000062	2013000062

 * <p>
 【开窗】窗口ID:[1577881919999]; 事件时间:[2020-01-01 20:31:40];结束窗口时间: [2020-01-01 20:31:59];司机ID:[2013000108] ;isStart:[false];
 【开窗】窗口ID:[1577881679999]; 事件时间:[2020-01-01 20:27:00];结束窗口时间: [2020-01-01 20:27:59];司机ID:[2013000134] ;isStart:[false];
 【开窗】窗口ID:[1577881799999]; 事件时间:[2020-01-01 20:29:20];结束窗口时间: [2020-01-01 20:29:59];司机ID:[2013000062] ;isStart:[false];
 【开窗】窗口ID:[1577884919999]; 事件时间:[2020-01-01 21:21:00];结束窗口时间: [2020-01-01 21:21:59];司机ID:[2013000108] ;isStart:[true];
 2> (2013000134,1577881679999,1)
 【触发窗口】窗口ID:[1577881679999],触发时间:[2020-01-01 20:27:59];触发求和计算结果:(2013000134,1577881679999,1)
 2> (2013000062,1577881799999,1)
 【触发窗口】窗口ID:[1577881799999],触发时间:[2020-01-01 20:29:59];触发求和计算结果:(2013000062,1577881799999,1)
 2> (2013000108,1577881919999,1)
 【触发窗口】窗口ID:[1577881919999],触发时间:[2020-01-01 20:31:59];触发求和计算结果:(2013000108,1577881919999,1)
 【开窗】窗口ID:[1577886419999]; 事件时间:[2020-01-01 21:46:00];结束窗口时间: [2020-01-01 21:46:59];司机ID:[2013000108] ;isStart:[false];
 2> (2013000108,1577884919999,1)
 【触发窗口】窗口ID:[1577884919999],触发时间:[2020-01-01 21:21:59];触发求和计算结果:(2013000108,1577884919999,1)
 【开窗】窗口ID:[1577885819999]; 事件时间:[2020-01-01 21:36:20];结束窗口时间: [2020-01-01 21:36:59];司机ID:[2013000062] ;isStart:[true];
 【开窗】窗口ID:[1577890739999]; 事件时间:[2020-01-01 22:58:20];结束窗口时间: [2020-01-01 22:58:59];司机ID:[2013000062] ;isStart:[false];
 2> (2013000062,1577885819999,1)
 【触发窗口】窗口ID:[1577885819999],触发时间:[2020-01-01 21:36:59];触发求和计算结果:(2013000062,1577885819999,1)
 2> (2013000108,1577886419999,1)
 【触发窗口】窗口ID:[1577886419999],触发时间:[2020-01-01 21:46:59];触发求和计算结果:(2013000108,1577886419999,1)
 2> (2013000062,1577890739999,1)
 【触发窗口】窗口ID:[1577890739999],触发时间:[2020-01-01 22:58:59];触发求和计算结果:(2013000062,1577890739999,1)
 */

public class Watermark02 {
    public static Long t1 = -9L;

    private final SourceFunction<TaxiRide> source;
    private static final OutputTag<String> rejectedWordsTag = new OutputTag<String>("rejected") {
    };

    public Watermark02(SourceFunction<TaxiRide> source) {
        this.source = source;
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<TaxiRide> rides = env.addSource(new TaxiRideGeneratorFromMysql());
        //水印策略
        WatermarkStrategy<TaxiRide> watermarkStrategy =
                WatermarkStrategy.<TaxiRide>forBoundedOutOfOrderness(Duration.ofMinutes(10))
                        .withTimestampAssigner(
                                (ride, streamRecordTimestamp) -> ride.getEventTimeMillis());//事件时间戳

        SingleOutputStreamOperator resCnt = rides
                .assignTimestampsAndWatermarks(watermarkStrategy)
                .keyBy((TaxiRide ride) -> ride.driverId)
                .process(new PseudoWindow(Time.minutes(1)));

        DataStream<Tuple3<Integer, Integer, Integer>> oneSideOutput = resCnt.getSideOutput(rejectedWordsTag);
//        oneSideOutput.print("【4】oneSideOutput");
        resCnt.print();

        env.execute();


    }

    public static class PseudoWindow extends KeyedProcessFunction<Long, TaxiRide, Tuple3<Long, Long, Long>> {
        private final long durationMsec;
        private transient MapState<Long, Long> driverIdCount;

        public PseudoWindow(Time durationMsec) {
            this.durationMsec = durationMsec.toMilliseconds();
        }

        //初始化
        @Override
        public void open(Configuration parameters) throws Exception {
            MapStateDescriptor<Long, Long> countDesc = new MapStateDescriptor<>("driverIdCount", Long.class, Long.class);
            driverIdCount = getRuntimeContext().getMapState(countDesc);
        }

        @Override
        public void processElement(TaxiRide ride, Context ctx, Collector<Tuple3<Long, Long, Long>> out) throws Exception {
            SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            long eventTime = ride.getEventTimeMillis();
            TimerService timerService = ctx.timerService();
//            System.out.println(String.format("【1】eventTime:%s currentWatermark:%s",eventTime, timerService.currentWatermark()));
            if (eventTime <= timerService.currentWatermark()) {
                ctx.output(rejectedWordsTag, ride.driverId+","+ride.rideId+","+ride.isStart);
            } else {
                //注册时间器
                long endOfWindow = (eventTime - (eventTime % durationMsec) + durationMsec - 1);
                timerService.registerEventTimeTimer(endOfWindow);
                Long count = driverIdCount.get(endOfWindow);
                //null值处理
                if (count == null) {
                    count = 0L;
                }

                System.out.printf("【开窗】窗口ID:[%s]; 事件时间:[%s];结束窗口时间: [%s];司机ID:[%s] ;isStart:[%s];\n",
                       endOfWindow, dateformat.format(eventTime), dateformat.format(endOfWindow), ride.driverId, ride.isStart);
                //driver count累加
                count+=1;
                driverIdCount.put(endOfWindow, count);
//                System.out.println(String.format("driverIdCount key:%s ,value:%s",endOfWindow,driverIdCount.get(endOfWindow)));
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple3<Long, Long, Long>> out) throws Exception {
            long driverId = ctx.getCurrentKey();
            Long count = this.driverIdCount.get(timestamp);
//            System.out.println("count:"+count);
            Tuple3<Long, Long, Long> result = Tuple3.of(driverId, timestamp, count);
            out.collect(result);
            SimpleDateFormat dateformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            System.out.printf("【触发窗口】窗口ID:[%s],触发时间:[%s];触发求和计算结果:%s \n", timestamp, dateformat.format(timestamp), result.toString());
            this.driverIdCount.remove(timestamp);
        }
    }
}


